package com.cbh.amadeus.client.utils;

import com.cbh.amadeus.client.config.SubscriberConfig;
import com.cbh.amadeus.client.domain.MessageMatchModel;
import com.cbh.amadeus.client.receiver.component.SubscriberComponent;
import com.cbh.amadeus.common.domain.AckModel;
import com.cbh.amadeus.common.enums.MessageType;
import com.cbh.amadeus.common.json.Json;
import com.cbh.amadeus.common.protocol.AmadeusMessage;
import com.cbh.amadeus.common.protocol.AmadeusMessageProtocol;
import com.cbh.amadeus.common.protocol.AmadeusMessageWrap;
import com.cbh.amadeus.common.utils.StringPool;
import com.cbh.amadeus.common.utils.StringUtil;
import io.netty.channel.Channel;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author ChenBiHui
 * @date 2025/4/28
 */
@Slf4j
@Data
public final class ChannelReference {

    private static final int LIMIT = 20;
    private static final List<Channel> CHANNEL = new CopyOnWriteArrayList<>();
    private static final Map<Channel, String> CHANNEL_WITH_NAME = new ConcurrentHashMap<>(64);
    private static final Map<String, MessageMatchModel> MESSAGE_WITH_MATCH = new ConcurrentHashMap<>(64);
    private static final AtomicBoolean BOOTING = new AtomicBoolean(true);
    private static final AtomicInteger COUNTER = new AtomicInteger(0);

    public static Channel get() {
        int count = 0;
        while (CHANNEL.isEmpty() && BOOTING.get()) {
            try {
                TimeUnit.MILLISECONDS.sleep(100);
                if (count++ == LIMIT) {
                    return null;
                }
            } catch (InterruptedException ignored) {
                return null;
            }
        }
        if (CHANNEL.isEmpty()) {
            return null;
        }
        ThreadLocalRandom current = ThreadLocalRandom.current();
        Channel channel = null;
        count = 0;
        do {
            try {
                channel = CHANNEL.get(current.nextInt(0, CHANNEL.size()));
                if (channel != null && channel.isActive()) {
                    if (BOOTING.get()) {
                        BOOTING.set(false);
                    }
                    break;
                }
                if (count++ == LIMIT) {
                    break;
                }
            } catch (IndexOutOfBoundsException e) {
                log.warn("===== ChannelReference#get() search channel's list has index out ou bound, can't get any channel for current call =====");
                break;
            }
        } while (true);
        return channel;
    }

    public static String findName(Channel channel) {
        String name = CHANNEL_WITH_NAME.get(channel);
        if (StringUtil.unHasText(name)) {
            return StringPool.EMPTY;
        }
        return name.substring(0, name.lastIndexOf(StringPool.DASH));
    }

    public static void set(final Channel channel, final String name) {
        CHANNEL.add(channel);
        CHANNEL_WITH_NAME.put(channel, name);
    }

    public static void remove(final Channel channel) {
        CHANNEL.remove(channel);
        CHANNEL_WITH_NAME.remove(channel);
    }

    public static Integer channelCount() {
        return COUNTER.get();
    }

    public static void loading(SubscriberComponent component) {
        SubscriberConfig config = component.getConfig();
        Integer channelCount = config.getChannelCount();
        if (channelCount == null || channelCount <= 0) {
            throw new IllegalArgumentException("The count must be greater than 0");
        }
        COUNTER.compareAndSet(0, channelCount);
    }

    public static String compose(String name, Integer sort) {
        if (StringUtil.unHasText(name)) {
            throw new IllegalArgumentException("The [SubscriberName] can't be empty, please set the prefix and ***[ensure global uniqueness]***");
        }
        return name + StringPool.DASH + sort;
    }

    public static void positioner(Channel channel, String name, AmadeusMessage message, Integer holders) {
        MESSAGE_WITH_MATCH.putIfAbsent(message.unique(), new MessageMatchModel(channel, name, message, holders));
    }

    public static MessageMatchModel seek(AmadeusMessage message) {
        return MESSAGE_WITH_MATCH.get(message.unique());
    }

    public static void ack(Channel channel, AmadeusMessage message, String name) {
        AmadeusMessageWrap wrap = new AmadeusMessageWrap();
        AmadeusMessage amadeus = new AmadeusMessage();
        AckModel ackModel = new AckModel();
        ackModel.setConsumer(name);
        ackModel.setAckTime(LocalDateTime.now());
        wrap.setMagic(AmadeusMessageProtocol.MAGIC);
        wrap.setType(MessageType.ACK.getType().byteValue());
        wrap.setMessage(amadeus);
        amadeus.setTenant(message.getTenant());
        amadeus.setProvider(message.getProvider());
        amadeus.setTopic(message.getTopic());
        amadeus.setMessageId(message.getMessageId());
        amadeus.setPayload(Json.payloadJson(ackModel));
        if (channel.isActive()) {
            channel.writeAndFlush(wrap)
                    .addListener(item -> {
                        if (item.isSuccess()) {
                            liberator(message);
                            log.info(
                                    "===== [SEND ACK] Topic [{}] message-id [{}] with [{}] is all finish =====",
                                    message.getTopic(),
                                    message.getMessageId(),
                                    name
                            );
                        }
                    });
        }
    }

    private static void liberator(AmadeusMessage message) {
        MESSAGE_WITH_MATCH.remove(message.unique());
    }
}
